#include <comm/atomic/atomic.h>
#include <comm/atomic/queue.h>
#include <comm/atomic/stack.h>
#include <comm/atomic/pool.h>
#include <comm/ref.h>
#include <comm/str.h>
#include <comm/pthreadx.h>
#include <comm/hptimer.h>
#include <comm/binstream/stdstream.h>

using namespace coid;

stdoutstream _out;

struct ref_test
{
	ref_test() : _wcycles(0), _rcycles(0) {}

	void print() {
		_out
			<< "read cycles: " << _rcycles << "\n"
			<< "write cycles: "	<< _wcycles << "\n";
	}

	volatile int32 _wcycles;
	volatile int32 _rcycles;
};

struct test 
{
	int32 _val;

	test() : _val(0) {}

	void reset() {}
};

typedef atomic::queue_ng<test> queue_t;
typedef atomic::stack<policy_queue_pooled<test>> pool_t;

queue_t _queue;
pool_t _pool;
const int POOL_SIZE = 10000000;
volatile int32 _items_written = 0;
volatile int32 _items_read = POOL_SIZE;
int32 _sum = 0;
int32 _sum2 = 0;
int32 _sum3 = 0;
ref<ref_test> _rtest;

static void* thread_writer( void* p )
{
	while (!thread::self_should_cancel() && _items_written < POOL_SIZE) {
		ref<test> item(policy_queue_pooled<test>::create());

		RASSERT(!item.is_empty());

		_sum3 += item->_val;

		//atomic::inc(&_items_written);
		++_items_written;

		_queue.push(item);

		//atomic::inc(&_rtest->_wcycles);
		++_rtest->_wcycles;
	}

	return 0;
}

static void* thread_reader( void* p )
{
	while (!thread::self_should_cancel() && _items_read > 0) {
		ref<test> item;

		if( _queue.pop(item) ) {
			_sum += item->_val;
			//atomic::dec(&_items_read);
			--_items_read;
		}
		//atomic::inc(&_rtest->_rcycles);
		++_rtest->_rcycles;
	}

	return 0;
}

void init_pool(const int size)
{
	coid::dynarray<ref<test>>  tmp;

	for (int i = 0; i < size; ++i) {
		(*tmp.add()).create(policy_queue_pooled<test>::create());
	}
}

int main_atomic(int argc, char * argv[])
{
	{
		typedef atomic::basic_pool<int> int_pool;

		int_pool p;

		int_pool::node_t* n=p.pop();

		n->_item=1234;

		p.push(n);

		n=p.pop();

		int a=0;
	}

	{
		struct test_oq {
			int a;
			int b;
			int c;
		};


		atomic::queue<test_oq> q;


	}


	const int num = 1;
	thread writers[num];
	thread readers[num];
	MsecTimer timer;

	_out.set_flush_token("");

	_rtest.create( new ref_test );

	{
		ref<ref_test> rtest( new ref_test );
	}

	init_pool(POOL_SIZE/1000);

	{
		_queue.push_take(ref<test>(policy_queue_pooled<test>::create()));
		_queue.push_take(ref<test>(policy_queue_pooled<test>::create()));
		ref<test> item1;
		RASSERT(_queue.pop(item1));
		ref<test> item2;
		RASSERT(_queue.pop(item2));
		ref<test> item3;
		RASSERT(!_queue.pop(item3));
	}

	_out << "pool initialized...\n";
	_out.flush();

	timer.reset();

	for( int i=0; i<num; ++i )
		writers[i].create( thread_writer,0,0,"writer" );
	sysMilliSecondSleep(100);
	for( int i=0; i<num; ++i )
		readers[i].create( thread_reader,0,0,"reader" );

	while( _items_read>0 ) {
		_out << "written: " << _items_written << " read: " << _items_read << "                 \r";
		_out.flush();
		sysMilliSecondSleep(10);
	}

	const int32 time = timer.time();

	_out << "written: " << _items_written << " read: " << _items_read << "                    \n";

	for (int i = 0; i < num; ++ i) {
		writers[i].cancel_and_wait(100);
		readers[i].cancel_and_wait(100);
	}

	_out << "\n"
		<< "time: " << static_cast<float>(time) / 1000.f << "\n"
		<< "items per second: " << POOL_SIZE / (static_cast<float>(time) / 1000.f) << "\n"
		<< "_items_read: " << _items_read << "\n"
		<< "_items_written: " << _items_written << "\n"
		<< "sum2: " << _sum2 << "\n"
		<< "sum:  " << _sum << "\n"
		<< "sum3:  " << _sum3 << "\n";
	
	_rtest->print();
	
	_out.flush();

	return 0;
}
